package com.yifeng.repo.apm.endpoint.sender;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yifeng.repo.apm.endpoint.console.api.LogConfService;
import com.yifeng.repo.apm.endpoint.console.api.model.LogConf;
import com.yifeng.repo.apm.endpoint.console.api.model.LogConfTypeEnum;
import com.yifeng.repo.apm.endpoint.console.proxy.LogConfServiceProxy;
import com.yifeng.repo.apm.endpoint.utils.BaseUtil;
import next.fire.spinus.logx.api.LogService;
import next.fire.spinus.logx.api.LogServiceFactory;
import next.fire.spinus.logx.api.model.LevelEnum;
import next.fire.spinus.logx.api.model.LogData;
import next.fire.spinus.logx.api.model.ProviderEnum;
import next.fire.spinus.logx.provider.aliyun.conf.AliyunLogConfig;
import next.fire.spinus.logx.provider.huawei.conf.HuaweiLogConfig;
import next.fire.spinus.logx.utils.async.AsyncMergeProcessor;
import next.fire.spinus.logx.utils.env.HostManager;
import next.fire.spinus.logx.utils.http.FireHttpUtils;

import java.io.*;
import java.net.SocketTimeoutException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by daibing on 2020/12/27.
 */
public class LogSender {
    private static final SimpleDateFormat SDF = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    private static final String ROOT_LOGGER_PREFIX = "log4j.rootLogger";
    private static final String LOGGER_PREFIX = "log4j.logger.";
    private static final String APPENDER_PREFIX = "log4j.appender.";
    private final ConcurrentMap<LogConfTypeEnum, Boolean> logStateMap = new ConcurrentHashMap<>(LogConfTypeEnum.values().length);
    private final AtomicInteger rootLoggerLevel = new AtomicInteger(Integer.MAX_VALUE);
    private final AtomicBoolean startupFinished = new AtomicBoolean(false);
    private final List<LogData> startupLogDataList = new ArrayList<>();
    private final String appId;
    private final String stage;
    private final LogConfService logConfService;
    private ConcurrentNavigableMap<String, LevelEnum> logFilterParams;
    private AsyncMergeProcessor<LogData> asyncMergeProcessor;
    private LogService logService;

    public LogSender(String appId, String stage, String logConfUrl) {
        this.appId = appId;
        this.stage = stage;
        this.logConfService = new LogConfServiceProxy(logConfUrl, null, 3 * 1000);
        for (LogConfTypeEnum type : LogConfTypeEnum.values()) {
            logStateMap.put(type, false);
        }
    }

    public void startup(boolean lastTime) {
        // 1. 加载日志远程配置
        List<LogConf> logConfList = this.safeListConf(appId, stage);
        if (logConfList == null || logConfList.size() == 0) {
            this.directPrintLog(LevelEnum.INFO, "LogSender-startup", String.format("ListConf is empty: appId=%s, stage=%s", appId, stage));
            return;
        }

        // 2. 从日志配置解析根日志级别、日志配置类型
        LevelEnum rootLoggerLevel = this.parseRootLoggerLevel(logConfList);
        this.rootLoggerLevel.set(rootLoggerLevel != null ? rootLoggerLevel.ordinal() : Integer.MAX_VALUE);
        List<LogConfTypeEnum> logConfTypeList = this.parseLogConfType(logConfList);
        for (LogConfTypeEnum type : logConfTypeList) {
            this.logStateMap.put(type, true);
        }

        // 3. 从日志配置解析日志过滤参数，如果是CLOUD就拉起日志中心服务和异步处理器
        this.logFilterParams = this.parseLogFilterParams(logConfList);
        if (!this.logStateMap.get(LogConfTypeEnum.CLOUD)) {
            return;
        }
        this.logService = this.buildLogService(logConfList, lastTime);
        this.asyncMergeProcessor = this.buildLogProcessor(logConfList);
        this.asyncMergeProcessor.startup();
        this.startupFinished.set(true);

        // 4. 将启动sender处理器前的日志推送到日志中心
        for (LogData logData : startupLogDataList) {
            this.asyncMergeProcessor.submit(logData);
        }
        startupLogDataList.clear();
        this.directPrintLog(LevelEnum.INFO, "LogSender-startup", String.format("start async merge processor ok: appId=%s, stage=%s", appId, stage));
    }

    public void shutdown() {
        if (asyncMergeProcessor != null) {
            asyncMergeProcessor.shutdown();
        }
    }

    public boolean matched(String className, LevelEnum level) {
        if (BaseUtil.isBlank(className) || level == null) {
            return false;
        }
        String str = LOGGER_PREFIX + className;
        Map.Entry<String, LevelEnum> floorEntry = logFilterParams.floorEntry(str);
        if (floorEntry == null && rootLoggerLevel.get() <= level.ordinal()) {
            return true;
        }
        return floorEntry != null && str.startsWith(floorEntry.getKey()) && floorEntry.getValue().ordinal() <= level.ordinal();
    }

    public void send(LogData logData) {
        if (!logStateMap.get(LogConfTypeEnum.CLOUD) || asyncMergeProcessor == null) {
            this.directPrintLog(LevelEnum.WARN, "LogSender-send", String.format("send failed by status error: state=%s, asyncMergeProcessor=%s", logStateMap.get(LogConfTypeEnum.CLOUD), asyncMergeProcessor));
            return;
        }
        asyncMergeProcessor.submit(logData);
    }

    public boolean isReady() {
        if (this.logFilterParams == null) {
            return false;
        }
        if (!logStateMap.get(LogConfTypeEnum.CLOUD)) {
            return true;
        }
        return logService != null && asyncMergeProcessor != null;
    }

    public boolean isLogEnabled(LogConfTypeEnum type) {
        return this.logStateMap.get(type);
    }

    public void refreshFilterParams() {
        List<LogConf> logConfList = safeListConf(appId, stage);
        if (logConfList == null || logConfList.size() == 0) {
            this.directPrintLog(LevelEnum.WARN, "LogSender-refresh-check-conf", String.format("ListConf is empty: appId=%s, stage=%s", appId, stage));
            return;
        }
        // 异步刷新仅支持更新根日志级别和日志过滤参数
        LevelEnum newRootLoggerLevel = parseRootLoggerLevel(logConfList);
        ConcurrentNavigableMap<String, LevelEnum> newLogFilterParams = parseLogFilterParams(logConfList);
        synchronized (rootLoggerLevel) {
            rootLoggerLevel.set(newRootLoggerLevel != null ? newRootLoggerLevel.ordinal() : Integer.MAX_VALUE);
            logFilterParams = newLogFilterParams;
        }
        this.directPrintLog(LevelEnum.INFO, "LogSender-refresh-ok",
                String.format("refreshFilterParams ok: appId=%s, stage=%s, rootLoggerLevel=%s, level=%s",
                        appId, stage, rootLoggerLevel.get(), LevelEnum.getByOrdinal(rootLoggerLevel.get())));
    }

    public void directPrintLog(LevelEnum level, String location, String message) {
        this.directPrintLog(level, location, message, null);
    }

    public void directPrintLog(LevelEnum level, String location, String message, Throwable t) {
        String errMsg = t == null ? "" : ", error=" + this.render(t);
        String body = String.format("%s [%s] %s directPrintLog: %s %s",
                SDF.format(System.currentTimeMillis()), Thread.currentThread().getName(), level.name(), location, message);
        if (level == LevelEnum.WARN || level == LevelEnum.ERROR) {
            System.err.println(body + errMsg);
        } else {
            System.out.println(body + errMsg);
        }
        if (this.startupFinished.get() && (!logStateMap.get(LogConfTypeEnum.CLOUD) || asyncMergeProcessor == null)) {
            return;
        }
        LogData logData = new LogData();
        logData.setAppId(this.appId);
        logData.setStage(this.stage);
        logData.setLevel(level);
        logData.setLogTime(System.currentTimeMillis());
        logData.setTraceId(null);
        logData.setFrontendId(null);
        logData.setHost(HostManager.get().getHostName());
        logData.setThread(Thread.currentThread().getName());
        logData.setLocation("directPrintLog: " + location);
        logData.setMessage(message);
        logData.setThrowable(t == null ? null : t.getMessage());
        if (this.startupFinished.get()) {
            asyncMergeProcessor.submit(logData);
        } else {
            startupLogDataList.add(logData);
        }
    }

    private List<LogConf> safeListConf(String appId, String stage) {
        try {
            return logConfService.list(appId, stage);
        } catch (Throwable t) {
            this.directPrintLog(LevelEnum.WARN, "LogSender-safeListConf", String.format("safeListConf failed: appId=%s, stage=%s", appId, stage), t);
            return null;
        }
    }

    private LevelEnum parseRootLoggerLevel(List<LogConf> logConfList) {
        for (LogConf logConf : logConfList) {
            if (LogConfTypeEnum.ROOT == logConf.getType() && ROOT_LOGGER_PREFIX.equalsIgnoreCase(logConf.getKey())) {
                String[] words = logConf.getValue().split(",");
                if (words.length > 1) {
                    this.directPrintLog(LevelEnum.INFO, "LogSender-parseRootLoggerLevel", String.format("parseRootLoggerLevel: level=%s, lastModify=%s", words[0].trim(), logConf.getLastModify()));
                    return LevelEnum.get(words[0].trim());
                }
                break;
            }
        }
        return null;
    }

    private List<LogConfTypeEnum> parseLogConfType(List<LogConf> logConfList) {
        for (LogConf logConf : logConfList) {
            if (LogConfTypeEnum.ROOT == logConf.getType() && ROOT_LOGGER_PREFIX.equalsIgnoreCase(logConf.getKey())) {
                String[] words = logConf.getValue().split(",");
                if (words.length == 1) {
                    return Collections.emptyList();
                }
                List<LogConfTypeEnum> logConfTypeList = new ArrayList<>();
                for (int i = 1; i < words.length; i++) {
                    logConfTypeList.add(LogConfTypeEnum.get(words[i].trim()));
                }
                return logConfTypeList;
            }
        }
        return Collections.emptyList();
    }

    private ConcurrentNavigableMap<String, LevelEnum> parseLogFilterParams(List<LogConf> logConfList) {
        ConcurrentNavigableMap<String, LevelEnum> logFilterParams = new ConcurrentSkipListMap<>();
        for (LogConf logConf : logConfList) {
            if (LogConfTypeEnum.ROOT == logConf.getType() && !ROOT_LOGGER_PREFIX.equalsIgnoreCase(logConf.getKey())) {
                logFilterParams.put(logConf.getKey(), LevelEnum.get(logConf.getValue()));
            }
        }
        return logFilterParams;
    }

    private Map<String, String> getParamsByPrefix(List<LogConf> logConfList, LogConfTypeEnum logConfType, String prefix) {
        Map<String, String> params = new HashMap<>();
        for (LogConf logConf : logConfList) {
            if (logConf.getType() == logConfType && logConf.getKey().toLowerCase().startsWith(prefix)) {
                params.put(logConf.getKey().toLowerCase().replace(prefix, ""), logConf.getValue());
            }
        }
        return params;
    }

    private LogService buildLogService(List<LogConf> logConfList, boolean lastTime) {
        String aliyunLogStoreCodeKey = APPENDER_PREFIX + LogConfTypeEnum.CLOUD.name() + "." + ProviderEnum.ALIYUN.name() + ".LogStoreCode";
        boolean aliyunLog = false;
        for (LogConf logConf : logConfList) {
            if (logConf.getKey().equalsIgnoreCase(aliyunLogStoreCodeKey)) {
                aliyunLog = true;
                break;
            }
        }
        if (aliyunLog) {
            return buildAliyunLogService(logConfList, lastTime);
        } else {
            return buildHuaweiLogService(logConfList);
        }
    }

    private LogService buildAliyunLogService(List<LogConf> logConfList, boolean lastTime) {
        String prefix = APPENDER_PREFIX + LogConfTypeEnum.CLOUD.name() + "." + ProviderEnum.ALIYUN.name() + ".";
        Map<String, String> aliyunLogParams = getParamsByPrefix(logConfList, LogConfTypeEnum.CLOUD, prefix.toLowerCase());
        String accountIntranetEndpoint = aliyunLogParams.get("AccountIntranetEndpoint".toLowerCase());
        String endpoint = this.checkConnection(accountIntranetEndpoint, 3 * 1000, lastTime) ? accountIntranetEndpoint : aliyunLogParams.get("AccountEndpoint".toLowerCase());
        AliyunLogConfig aliyunLogConfig = new AliyunLogConfig();
        aliyunLogConfig.setLogStoreCode(aliyunLogParams.get("LogStoreCode".toLowerCase()));
        aliyunLogConfig.setAccessKeyId(aliyunLogParams.get("AccessKeyId".toLowerCase()));
        aliyunLogConfig.setAccessKeySecret(aliyunLogParams.get("AccessKeySecret".toLowerCase()));
        aliyunLogConfig.setAccountEndpoint(endpoint);
        aliyunLogConfig.setLogProject(aliyunLogParams.get("LogProject".toLowerCase()));
        aliyunLogConfig.setLogStore(aliyunLogParams.get("LogStore".toLowerCase()));
        LogServiceFactory.get(ProviderEnum.ALIYUN).loadConf((JSONObject) JSON.toJSON(aliyunLogConfig));
        return LogServiceFactory.get(ProviderEnum.ALIYUN);
    }

    private LogService buildHuaweiLogService(List<LogConf> logConfList) {
        String prefix = APPENDER_PREFIX + LogConfTypeEnum.CLOUD.name() + "." + ProviderEnum.HUAWEI.name() + ".";
        Map<String, String> huaweiLogParams = getParamsByPrefix(logConfList, LogConfTypeEnum.CLOUD, prefix.toLowerCase());
        HuaweiLogConfig huaweiLogConfig = new HuaweiLogConfig();
        huaweiLogConfig.setLogStoreCode(huaweiLogParams.get("LogStoreCode".toLowerCase()));
        huaweiLogConfig.setAccessKeyId(huaweiLogParams.get("AccessKeyId".toLowerCase()));
        huaweiLogConfig.setAccessKeySecret(huaweiLogParams.get("AccessKeySecret".toLowerCase()));
        huaweiLogConfig.setAccountEndpoint(huaweiLogParams.get("AccountEndpoint".toLowerCase()));
        huaweiLogConfig.setAgentEndpoint(huaweiLogParams.get("AgentEndpoint".toLowerCase()));
        huaweiLogConfig.setProjectId(huaweiLogParams.get("ProjectId".toLowerCase()));
        //huaweiLogConfig.setLogGroupName(huaweiLogParams.get("LogGroupName".toLowerCase()));
        huaweiLogConfig.setLogGroupId(huaweiLogParams.get("LogGroupId".toLowerCase()));
        //huaweiLogConfig.setLogStreamName(huaweiLogParams.get("LogStreamName".toLowerCase()));
        huaweiLogConfig.setLogStreamId(huaweiLogParams.get("LogStreamId".toLowerCase()));
        huaweiLogConfig.setAccountName(huaweiLogParams.get("AccountName".toLowerCase()));
        huaweiLogConfig.setIamEndpoint(huaweiLogParams.get("IamEndpoint".toLowerCase()));
        huaweiLogConfig.setIamUserName(huaweiLogParams.get("IamUserName".toLowerCase()));
        huaweiLogConfig.setIamPassword(huaweiLogParams.get("IamPassword".toLowerCase()));
        LogServiceFactory.get(ProviderEnum.HUAWEI).loadConf((JSONObject) JSON.toJSON(huaweiLogConfig));
        return LogServiceFactory.get(ProviderEnum.HUAWEI);
    }

    private AsyncMergeProcessor<LogData> buildLogProcessor(List<LogConf> logConfList) {
        // 当时为支持tomcat日志设置两种类型（tomcat、app），当前仅保留app，所以直接固定带上.app
        String prefix = APPENDER_PREFIX + LogConfTypeEnum.CLOUD.name().toLowerCase() + ".sender.app.";
        Map<String, String> senderParams = getParamsByPrefix(logConfList, LogConfTypeEnum.CLOUD, prefix);
        return new AsyncMergeProcessor<LogData>("LOGX", Integer.parseInt(senderParams.getOrDefault("ThreadSize".toLowerCase(), "1")), Integer.parseInt(senderParams.getOrDefault("BufferSize".toLowerCase(), "1024"))) {
            final AtomicInteger count = new AtomicInteger(0);

            @Override
            protected int pageSize() {
                return Integer.parseInt(senderParams.getOrDefault("PageSize".toLowerCase(), "100"));
            }

            @Override
            protected int queuePollTimeoutMillis() {
                return 50;
            }

            @Override
            protected int handlePageData(List<LogData> itemList) {
                if (itemList == null || itemList.isEmpty()) {
                    return 0;
                }
                logService.putData(senderParams.getOrDefault("LogStoreCode".toLowerCase(), "LOGX"), itemList);
                return itemList.size();
            }

            @Override
            protected void handlePageErrorData(List<LogData> itemList, Throwable t) {
                int size = itemList == null ? 0 : itemList.size();
                directPrintLog(LevelEnum.WARN, "LogSender-handlePageErrorData", String.format("logx send failed: itemList.size=%s", size), t);
            }

            @Override
            protected void print(int queueSize, int itemsSize, int handleCount) {
                if (count.incrementAndGet() >= 1500) {
                    directPrintLog(LevelEnum.INFO, "LogSender-print", String.format("logx send ok: queue.size=%s, items.size=%s, count=%s", queueSize, itemsSize, handleCount));
                    count.set(0);
                }
            }
        };
    }

    private boolean checkConnection(String endPoint, int timeoutMillis, boolean lastTime) {
        if (BaseUtil.isBlank(endPoint)) {
            return false;
        }
        try {
            FireHttpUtils httpClient = new FireHttpUtils(endPoint, timeoutMillis, true);
            Map<String, String> result = httpClient.get();
            this.directPrintLog(LevelEnum.INFO, "LogSender-checkConnection-check", String.format("check connection is ok: endPoint=%s, result=%s", endPoint, result));
            return true;
        } catch (Throwable t) {
            if (lastTime) {
                if (t instanceof SocketTimeoutException) {
                    this.directPrintLog(LevelEnum.WARN, "LogSender-checkConnection-timeout", String.format("check connection is not ok: endPoint=%s", endPoint), t);
                    return false;
                } else {
                    this.directPrintLog(LevelEnum.WARN, "LogSender-checkConnection-ok", String.format("check connection is not ok: endPoint=%s", endPoint), t);
                    return true;
                }
            } else {
                if (t instanceof SocketTimeoutException) {
                    this.directPrintLog(LevelEnum.INFO, "LogSender-checkConnection-timeout", String.format("check connection is not ok: endPoint=%s, msg=%s", endPoint, t.getMessage()));
                    return false;
                } else {
                    this.directPrintLog(LevelEnum.INFO, "LogSender-checkConnection-ok", String.format("check connection is not ok: endPoint=%s, msg=%s", endPoint, t.getMessage()));
                    return true;
                }
            }
        }
    }

    private String render(Throwable t) {
        StringWriter sw = new StringWriter();
        PrintWriter pw = new PrintWriter(sw);
        try {
            t.printStackTrace(pw);
        } catch (RuntimeException e) {
            // ignore.
        }
        pw.flush();
        LineNumberReader reader = new LineNumberReader(new StringReader(sw.toString()));
        List<String> lines = new ArrayList<>();
        try {
            String line = reader.readLine();
            while (line != null) {
                lines.add(line);
                line = reader.readLine();
            }
        } catch (IOException e) {
            if (e instanceof InterruptedIOException) {
                Thread.currentThread().interrupt();
            }
            lines.add(e.toString());
        }
        return JSON.toJSONString(lines);
    }

}
